package com.mjf.source

import java.util.UUID

import com.mjf.dim.MarketUserBehavior
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}

import scala.util.Random

/**
 * 自定义测试数据源
 */
class SimulateMarketEventSource extends RichParallelSourceFunction[MarketUserBehavior]{

  // 定义是否在运行的标志位
  private var running: Boolean = true

  // 定义用户行为和推广渠道的集合
  private val behaviorSet: Seq[String] = Seq("CLICK", "DOWNLOAD", "INSTALL", "UNINSTALL")
  private val channelSet: Seq[String] = Seq("appStore", "huaweiStore", "weibo", "wechat")

  // 定义随机数生成器
  private val rand: Random = Random

  override def run(ctx: SourceFunction.SourceContext[MarketUserBehavior]): Unit = {
    // 定义一个发出数据的最大量，用于控制测试数据量
    val maxCount: Long = Long.MaxValue
    var count: Long = 0L

    // while循环，不停的随机生成数据
    while(running && count < maxCount) {
      val id: String = UUID.randomUUID().toString
      val behavior: String = behaviorSet(rand.nextInt(behaviorSet.size))
      val channel: String = channelSet(rand.nextInt(channelSet.size))
      val ts: Long = System.currentTimeMillis()

      ctx.collect(MarketUserBehavior(id, behavior, channel, ts))

      count += 1
      Thread.sleep(50L)
    }
  }

  override def cancel(): Unit = running = false
}
